package com.chukun.disruptor.netty.disruptor.factory;

import com.chukun.disruptor.netty.disruptor.consumer.MessageConsumer;
import com.chukun.disruptor.netty.disruptor.entity.TranslatorDataWrapper;
import com.chukun.disruptor.netty.disruptor.producer.MessageProducer;
import com.lmax.disruptor.*;
import com.lmax.disruptor.dsl.ProducerType;

import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 创建RingBuffer工作池工厂
 */
public class RingBufferWorkerPoolFactory {


    private static Map<String, MessageProducer> producerMap = new ConcurrentHashMap<>();
    private static Map<String, MessageConsumer> consumerMap = new ConcurrentHashMap<>();

    private  RingBuffer<TranslatorDataWrapper> ringBuffer;
    private SequenceBarrier sequenceBarrier;
    private WorkerPool<TranslatorDataWrapper> workerPool;

    private ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(Runtime.getRuntime().availableProcessors(),Runtime.getRuntime().availableProcessors()+1,60, TimeUnit.SECONDS,new ArrayBlockingQueue<>(10));

    public void initAndStart(ProducerType type, int bufferSize, WaitStrategy waitStrategy, MessageConsumer[] messageConsumers){
        //1.构建ringBuffer实例
        this.ringBuffer = RingBuffer.create(type, new EventFactory<TranslatorDataWrapper>() {
            @Override
            public TranslatorDataWrapper newInstance() {
                return new TranslatorDataWrapper();
            }
        },bufferSize,waitStrategy);

        //2.设置序号栅栏
        this.sequenceBarrier = this.ringBuffer.newBarrier();

        //3.设置工作池
        this.workerPool = new WorkerPool<>(this.ringBuffer,this.sequenceBarrier,new EventExceptionHandler(),messageConsumers);

        //4.将consumers放入池中
        for (MessageConsumer mc: messageConsumers){
            consumerMap.put(mc.getConsumerID(),mc);
        }
        //5.添加sequences
        this.ringBuffer.addGatingSequences(this.workerPool.getWorkerSequences());

        //6.启动workpool
        this.workerPool.start(threadPoolExecutor);

    }

    /**
     * 获取生产者的producer
     * @param producerID
     * @return
     */
    public MessageProducer getMessageProducer(String producerID){
        if(producerMap.containsKey(producerID)){
            return producerMap.get(producerID);
        }else {
            MessageProducer messageProducer = new MessageProducer(producerID,this.ringBuffer);
            producerMap.put(producerID,messageProducer);
            return messageProducer;
        }
    }


    /**
     * 定义单例类
     */
    private static class SingletonHolder{
        static final RingBufferWorkerPoolFactory  instance = new RingBufferWorkerPoolFactory();
    }

    public static RingBufferWorkerPoolFactory getInstance(){
        return SingletonHolder.instance;
    }

    static class EventExceptionHandler implements ExceptionHandler<TranslatorDataWrapper>{

        @Override
        public void handleEventException(Throwable ex, long sequence, TranslatorDataWrapper event) {

        }

        @Override
        public void handleOnStartException(Throwable ex) {

        }

        @Override
        public void handleOnShutdownException(Throwable ex) {

        }
    }

}
